package com.github.shiqyue.timedelay;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/***
 * 由wwy创建于2018/5/30
 */
public class RedisTimeDelayManager implements TimeDelayManager, Runnable {

    private static final Logger log = LoggerFactory.getLogger(RedisTimeDelayManager.class);

    protected RedisTemplate redisTemplate;

    /***
     * 键值前缀
     */
    protected String keyPrefix = "timedelay";

    /***
     * 消费者List
     */
    protected List<TimeDelayConsumer> consumerList = new ArrayList<>(10);

    /***
     * 在redis中保存现在注册的type，对应的键值
     */
    private String typesKey = "";


    public RedisTimeDelayManager() {
        typesKey = keyPrefix + ":types";
    }

    public RedisTimeDelayManager(RedisTemplate redisTemplate) {
        this();
        Assert.notNull(redisTemplate, "redisTemplate不能为空");
        this.redisTemplate = redisTemplate;
    }

    public RedisTimeDelayManager(RedisTemplate redisTemplate, String keyPrefix) {
        this(redisTemplate);
        Assert.hasText(keyPrefix, "keyPrefix不能为空");
        this.keyPrefix = keyPrefix;
    }

    @Override
    public boolean addTask(String type, String id, Long timeDelay, TimeUnit timeUnit) {
        try {
            if (timeDelay <= 0) {
                //如果延迟时间小于0，则直接返回false
                return false;
            }
            timeDelay = timeUnit.toMillis(timeDelay);
            Long currentTime = System.currentTimeMillis();
            String key = keyPrefix + ":" + type;
            redisTemplate.opsForZSet().add(key, id, currentTime + timeDelay);
            redisTemplate.opsForSet().add(typesKey, type);
            return true;
        } catch (Exception ex) {
            ex.printStackTrace();
            return false;
        }
    }

    @Override
    public void addConsumer(TimeDelayConsumer timeDelayConsumer) {
        consumerList.add(timeDelayConsumer);
    }

    @Override
    public void consume() {
        Thread consumeThread = new Thread(this);
        consumeThread.start();
    }


    @Override
    public void run() {
        while (true) {
            try {
                Set<String> types = redisTemplate.opsForSet().members(typesKey);
                if (types == null || types.isEmpty()) {
                    Thread.sleep(500);
                    continue;
                }
                Long currentTime = System.currentTimeMillis();
                for (String type : types) {
                    String key = keyPrefix + ":" + type;
                    Set<ZSetOperations.TypedTuple<String>> items = redisTemplate.opsForZSet().rangeByScoreWithScores(key, 0, currentTime);
                    if (items == null || items.isEmpty()) {
                        continue;
                    }
                    for (TimeDelayConsumer consumer : consumerList) {
                        if (!consumer.support(type)) {
                            continue;
                        }
                        for (ZSetOperations.TypedTuple<String> item : items) {
                            boolean removeSuccess = redisTemplate.opsForZSet().remove(key, item.getValue()) == 1;
                            if (!removeSuccess) {
                                continue;
                            }
                            try {
                                consumer.consume(item.getValue(), type);
                            } catch (Exception e) {
                                //消费延时任务失败，重新投递
                                redisTemplate.opsForZSet().add(key, item.getValue(), item.getScore());
                                log.error("消费延时任务异常", e);
                                e.printStackTrace();
                            }

                        }
                        break;
                    }
                }
                Thread.sleep(500);
            } catch (InterruptedException e) {
                log.error("消费延时任务异常", e);
                e.printStackTrace();
            }
        }
    }
}
